package com.youbug.kefu.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.youbug.common.constant.AgentConstant;
import com.youbug.common.constant.ChatConstant;
import com.youbug.common.constant.DicTypeContent;
import com.youbug.kefu.dto.AgentSessionInfo;
import com.youbug.kefu.dto.QueueInfo;
import com.youbug.kefu.dto.SessionInfo;
import com.youbug.kefu.entity.Chat;
import com.youbug.system.entity.Dictionary;
import com.youbug.kefu.entity.Message;
import com.youbug.kefu.service.*;
import com.youbug.kefu.websocket.ChatsInfoHolder;
import com.youbug.system.service.IDictionaryService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.event.ApplicationStartedEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.stereotype.Service;

import javax.websocket.Session;
import java.io.IOException;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

@Service
@Slf4j
public class QueueService implements IQueueService, ApplicationListener<ApplicationStartedEvent> {

    @Autowired
    private IChatService chatService;
    @Autowired
    private IMessageService messageService;
    @Autowired
    private ISenderService senderService;
    @Autowired
    private ObjectMapper objectMapper;
    @Autowired
    private IDictionaryService dictionaryService;

    private BlockingQueue<Chat> queue  = new ArrayBlockingQueue<>(1000);

    private final Object lock = new Object();

    private volatile static boolean GETTING  = true;
    private static AtomicInteger round  = new AtomicInteger(0);

    @Override
    public boolean toQueueUp(Chat chat) {
        queue.add(chat);
        broadcastQueueInfo();
        return true;
    }

    public void broadcastQueueInfo() {
        List<Dictionary> list = dictionaryService.listByTypeCode(DicTypeContent.CHAT_TYPE);
        Map<String, QueueInfo> queueInfos = new HashMap<>(list.size());
        list.forEach(dictionary -> {
            queueInfos.put(dictionary.getCode(), new QueueInfo(dictionary.getCode(), dictionary.getValue(), 0));
        });

        queue.forEach(chat -> {
            QueueInfo queueInfo = queueInfos.get(chat.getType());
            Integer waitingNum = queueInfo.getWaitingNum();
            queueInfo.setWaitingNum(waitingNum + 1);
        });

        ChatsInfoHolder.agentSessionInfoMap.forEach((agentId, agentSessionInfo) -> {
            Session session = agentSessionInfo.getSession();
            if (session.isOpen()) {
                try {
                    Message message = new Message();
                    message.setType(ChatConstant.MSG_TYPE_QUEUE);

                    Collection<QueueInfo> qinfos = queueInfos.values();

                    message.setContent(objectMapper.writeValueAsString(qinfos));

                    session.getBasicRemote().sendText(objectMapper.writeValueAsString(message));
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        });
    }

    @Override
    public void stopWaiting(Chat holdChat) {
        synchronized (lock) {
            queue.removeIf(chat -> chat.getId().equals(holdChat.getId()));
        }
    }

    @Override
    public boolean restartDistribute() {
        if(!GETTING){
            synchronized (lock) {
                GETTING = true;
                lock.notify();
            }
        }
        return true;
    }

    @Override
    public void onApplicationEvent(ApplicationStartedEvent event) {
        new Thread(() -> {
            log.info("start distribute chat。。。");
            while(true){
                log.info("一轮分配开始");
                synchronized (lock){
                    int size = ChatsInfoHolder.agentSessionInfoMap.size();

                    if(size > 0){
                    ChatsInfoHolder.agentSessionInfoMap.forEach((agentId, agentSessionInfo) -> {
                        Chat chat = null;
                        try {
                            log.info("从队列中获取开始");
                            chat = queue.take();
                            log.info("从队列中获取到会话");
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        boolean distribute = false;
                        if (agentSessionInfo.getSession().isOpen()
                                && AgentConstant.READY.getCode().equals(agentSessionInfo.getState())
                                && agentSessionInfo.getHoldChats().size() < agentSessionInfo.getMax()) {
                            chat.setAgentId(agentSessionInfo.getUserId());
                            chat.setAgentName(agentSessionInfo.getUserName());
                            chat.setState(ChatConstant.CHAT_STATE_TALKING);
                            chat.setAnswerTime(LocalDateTime.now());
                            chatService.updateById(chat);

                            agentSessionInfo.getHoldChats().put(chat.getId(), chat);

                            SessionInfo sessionInfo = ChatsInfoHolder.clientSessionInfoMap.get(chat.getClientId());
                            sessionInfo.setHoldChat(chat);

                            Message message = new Message(ChatConstant.SYSTEM, ChatConstant.MSG_TYPE_TEXT, chat.getAgentId()+"号客服人员，为您服务");
                            senderService.sendMsgToClient(chat.getClientId(), message);

                            distribute = true;
                            round.set(0);
                            sendMsgInWaiting(chat, agentSessionInfo);
                            broadcastQueueInfo();
                        }
                        if(!distribute){
                            round.incrementAndGet();
                            queue.add(chat);
                            if(round.get() == 2 * queue.size() * size){
                                try {
                                    GETTING = false;
                                    broadcastQueueInfo();
                                    lock.wait();
                                } catch (InterruptedException e) {
                                    e.printStackTrace();
                                }
                            }
                        }
                        });
                        log.info("distribute chat。。。");

                }else{
                  try {
                      GETTING = false;

                      lock.wait();

                      log.info("restart circle...");
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                  }
              }
                log.info("一轮分配结束");
            }
        },"distributeChatThread").start();
    }

    private void sendMsgInWaiting(Chat chat, AgentSessionInfo agentSessionInfo) {

        QueryWrapper<Message> wrapper = new QueryWrapper<>();
        wrapper.eq("chat_id", chat.getId());
        wrapper.eq("sender", ChatConstant.CLIENT);
        wrapper.eq("sender_id", chat.getClientId());

        List<Message> list = messageService.list(wrapper);

        list.forEach(msg->{
            try {
                agentSessionInfo.getSession().getBasicRemote().sendText(objectMapper.writeValueAsString(msg));
            } catch (IOException e) {
                e.printStackTrace();
            }
        });
    }
}
